package com.tairanchina.md.elife.rabbit.client.consumer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.tairanchina.md.elife.rabbit.client.RabbitClient;
import com.tairanchina.md.elife.rabbit.task.RabbitMQTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**  消费者客户端
 * Created by lxliuxuan on 2016/12/12.
 */
public abstract class RabbitConsumerClient extends RabbitClient {


    private Logger logger = LoggerFactory.getLogger(RabbitConsumerClient.class);
    /**
     *  接收消息
     *  @param queueName  队列名称
     *  @param num  每次消费的消息个数
     *  @param objs 需要透传的处理对象
     *  @throws IOException
     *  @throws InterruptedException
     */
    public  abstract  void receiveMessage(String queueName,int num,RabbitMQTask rabbitMQTask, Object... objs) throws
                                                                                                        Exception;

    public  abstract  void receiveMessage(String queueName,RabbitMQTask rabbitMQTask, Object... objs) throws Exception;

    public  void receiveMessageByJson(String queueName,int num,RabbitMQTask rabbitMQTask, Object... objs) throws Exception{
        Channel channel = null;
        try
        {
            //判断连接是否可用，如果不可用，就创建新的连接
            channel = getChannel();

            //设置每个消费者每次接收的消息数
            channel.basicQos(num);
            QueueingConsumer consumer = new QueueingConsumer(channel);

            //false 表示手动控制应答
            channel.basicConsume(queueName, false, consumer);
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();//得到交付
            String str = new String(delivery.getBody());


            //处理业务逻辑
            rabbitMQTask.consumerMsg(str, objs);

            //确认消息已经收到
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
        catch (Exception e)
        {
            e.printStackTrace();
            logger.error("消息接收失败。e:{}", e);
            throw new Exception(e);
        }
        finally
        {
            try
            {
                if (channel != null)
                {
                    channel.close();
                }
            }
            catch (TimeoutException | IOException e)
            {
                e.printStackTrace();
                logger.error("消费者端消息通道关闭失败。e:{}", e);
            }
        }
    }
    public   void receiveMessageByJson(String queueName,RabbitMQTask rabbitMQTask, Object... objs) throws Exception{
        receiveMessageByJson(queueName,1,rabbitMQTask,objs);
    };
}
